package com.cwj.jdk;

import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author wujie.chen
 * 2022/11/28 9:50
 */
@Slf4j
public class Sync {
    private final Map<String, DelayQueue<TaimeiSyncEventDelayed<String>>> queues = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduledExecutorService;
    private final long delay;

    public Sync(long delay) {
        this.scheduledExecutorService = new ScheduledThreadPoolExecutor(10, defaultThreadFactory());
        this.delay = delay;
    }

    public void sync(String tenantId, Callable<Boolean> callable, long time) {
        DelayQueue<TaimeiSyncEventDelayed<String>> previous = queues.putIfAbsent(tenantId, new DelayQueue<>());
        if (previous == null) {
            System.out.println("启动消费者：" + tenantId);
            scheduledExecutorService.scheduleWithFixedDelay(() -> consume(tenantId, callable), delay, delay, TimeUnit.NANOSECONDS);
        }
        DelayQueue<TaimeiSyncEventDelayed<String>> queue = queues.get(tenantId);
        queue.put(new TaimeiSyncEventDelayed<>(null, time + delay, tenantId));
    }

    public boolean shutdown(long time, TimeUnit timeUnit) throws InterruptedException {
        ExecutorService executorService = (ExecutorService)scheduledExecutorService;
        return executorService.awaitTermination(time, timeUnit);

    }

    private void consume(String tenantId, Callable<Boolean> callable) {
        DelayQueue<TaimeiSyncEventDelayed<String>> queue = queues.get(tenantId);
        if (queue == null) {
            throw new RuntimeException("并发创建消费者失败");
        }
        List<TaimeiSyncEventDelayed<String>> list = new ArrayList<>();
        queue.drainTo(list);
        System.out.println(Thread.currentThread().getName() + " drainTo list size: " + list.size());

        TaimeiSyncEventDelayed<String> timer;
        if (!list.isEmpty()) {
            timer = list.get(list.size() - 1);
            System.out.println("now time: " + System.nanoTime() + " " +Thread.currentThread().getName() + "  " + timer);
            Task<String, ?> stringTask = new Task<>(timer, callable);
            stringTask.run();
        }
    }

    static class Task<T, R> implements Runnable {
        private final TaimeiSyncEventDelayed<T> timer;
        private final Callable<R> target;

        public Task(TaimeiSyncEventDelayed<T> timer, Callable<R> target) {
            this.timer = timer;
            this.target = target;
        }

        @Override
        public void run() {
            R result = null;
            try {
                result = target.call();
            } catch (Exception exception) {
                exception.printStackTrace();
            }
            log.info("timer task... data: {}, name: {}, time: {} result is: {}", timer.t, timer.name,
                TimeUnit.NANOSECONDS.toSeconds(timer.time), result);
        }
    }

    static class TaimeiSyncEventDelayed<T> implements Delayed {
        private final String name;
        private final long time;
        private final T t;

        public TaimeiSyncEventDelayed(String name, long time, T t) {
            this.name = name;
            this.time = time;
            this.t = t;
        }

        @Override
        public String toString() {
            return "Timer{" + "name='" + name + '\'' + ", time=" + time + ", t=" + t + '}';
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(time - System.nanoTime(), TimeUnit.NANOSECONDS);
        }

        @Override
        public int compareTo(@NonNull Delayed other) {
            if (other == this) { // compare zero ONLY if same object
                return 0;
            }
            if (other instanceof TaimeiSyncEventDelayed) {
                TaimeiSyncEventDelayed<?> x = (TaimeiSyncEventDelayed<?>)other;
                long diff = time - x.time;
                if (diff < 0) {
                    return -1;
                } else if (diff > 0) {
                    return 1;
                } else {
                    return 0;
                }
            }
            long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS));
            return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
        }
    }

    public static ThreadFactory defaultThreadFactory() {
        return new SyncTaimeiThreadFactory();
    }

    private static class SyncTaimeiThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        SyncTaimeiThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = "sync-taimei-pool-" + poolNumber.getAndIncrement() + "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (t.isDaemon()) {
                t.setDaemon(false);
            }
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }
}
